package kafka_day02

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.kafka.common.TopicPartition

import scala.collection.mutable

object MyUtils {
  def queryHistoryOffetFromMySQL(appNameAndGroupId: String): Map[TopicPartition, Long] = {

    val offsets: mutable.HashMap[TopicPartition, Long] = new mutable.HashMap[TopicPartition, Long]()

    //建立数据库的链接


    var conn: Connection = null
    var preparedStatement: PreparedStatement = null

    try {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/guanxi?characterEncoding=utf-8", "root", "123456")
      preparedStatement = conn.prepareStatement("SELECT topic_partition, offset FROM t_kafka_offset WHERE app_gid = ?")

      preparedStatement.setString(1, appNameAndGroupId)

      val resultSet = preparedStatement.executeQuery()

      //查询历史偏移量
      while (resultSet.next()) {

        val topicAndPartition = resultSet.getString("topic_partition")
        val fields = topicAndPartition.split("_")
        val offset = resultSet.getLong("offset")

        val topicPartition = new TopicPartition(fields(0), fields(1).toInt)

        offsets += ((topicPartition, offset))
      }
    } catch {
      case e =>{
        throw new RuntimeException("读取历史偏移量信息失败", e)
      }
    } finally {
     if(preparedStatement!=null){
       preparedStatement.close()
     }
      if(conn!=null){
        conn.close()
      }

    }


//返回Map
return offsets.toMap
  }


}
